package org.apache.flink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.endpoint.ResourceManagerRpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcSystem;
import org.apache.flink.shaded.curator5.org.apache.curator.RetryPolicy;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.flink.shaded.curator5.org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.CreateMode;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

public class ResourceManager implements LeaderLatchListener {


    private static String RM_LEADER_PATH = "/flink_rm_leader";
    public static String RM_LEADER_ADDRESS = "/flink_rm_address";

    private boolean isRunning = false;
    private String address = "";

    private CuratorFramework client;

    public void startService(String port) throws Exception {
        Configuration configuration = new Configuration();
        RpcSystem rpcSystem = RpcSystem.load(configuration);
        RpcService rpcService = rpcSystem.remoteServiceBuilder(configuration, "127.0.0.1", port)
                .createAndStart();

        ResourceManagerRpcEndpoint resourceManagerRpcEndpoint = new ResourceManagerRpcEndpoint(rpcService, "rm_" + UUID.randomUUID().toString());
        resourceManagerRpcEndpoint.start();
        this.address = resourceManagerRpcEndpoint.getAddress();

        isRunning = true;

        // 选主
        client = CuratorFrameworkFactory.builder()
                .connectString("172.30.18.45:2181")
                .retryPolicy(new ExponentialBackoffRetry(3, 4))
                .namespace("flink")
                .build();
        client.start();

        LeaderLatch leaderLatch = new LeaderLatch(client, RM_LEADER_PATH);
        leaderLatch.addListener(this);
        leaderLatch.start();
    }


    @Override
    public void isLeader() {
        try {
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(RM_LEADER_ADDRESS, this.address.getBytes(StandardCharsets.UTF_8));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void notLeader() {
        try {
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(RM_LEADER_ADDRESS, "".getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void main(String[] args) throws Exception {
        System.out.println("ResourceManager start...");

        ResourceManager resourceManager = new ResourceManager();
        resourceManager.startService(args[0]);
    }
}
